package com.yanggu.flink.datastream_api.test

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.common.typeinfo.Types
import org.apache.flink.util.Collector

/**
 * FlatMap函数使用了状态
 */
class StatefulFlatMapFunction extends RichFlatMapFunction[Integer, Integer] {

  // cumulative value
  private lazy val total = getRuntimeContext.getState(new ValueStateDescriptor[Integer]("totalValue", Types.INT, 0))

  @throws[Exception]
  override def flatMap(value: Integer, out: Collector[Integer]): Unit = {
    val value2 = value + total.value()
    total.update(value2)
    out.collect(value2)
  }

}